package com.shikong.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

public class kafkaProducerTest {

    public static void main(String[] args) {

        // TODO 创建配置对象
        Map<String, Object> configMap = new HashMap<String, Object>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        //TODO 对生产的数据K,V进行序列化
        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // TODO 创建生产者对象
        //    1、生产者对象需要设定泛型：数据类型的约束
        KafkaProducer<String,String> producer = new KafkaProducer<String,String>(configMap);

        // TODO 创建数据
        //     构建数据需要传3个参数
        //     1、topic 表述主题的名臣
        //     2、key 表述数据的key
        //     3、value 表述数据的value
      //  ProducerRecord<String,String> record = new ProducerRecord<String,String>(
     //           "test","hello","world");

        // TODO 通过生产者对象将数据发送到kafka
     //   producer.send(record);

        for (int i = 0; i < 10; i++) {
            ProducerRecord<String,String> record = new ProducerRecord<String,String>(
                    "test","hello"+i,"world"+i);
            // TODO 通过生产者对象将数据发送到kafka
            producer.send(record);
        }

        //TODO 关闭生产者对象
        producer.close();
    }
}
